/**
 *    Copyright 2009 Seth Albnaese
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.aquiferserver.server;

import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipelineCoverage;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.util.internal.MapBackedSet;

/**
 * @author Seth Albanese
 * @version $Id: AquiferServerHandler.java 42 2009-05-13 15:45:10Z salbanese0080@gmail.com $
 */

@ChannelPipelineCoverage( "all" )
public class AquiferServerHandler extends SimpleChannelHandler 
{
	protected static final Log logger = LogFactory.getLog( org.aquiferserver.server.AquiferServerHandler.class );
	
	static final Set<Channel> channels = new MapBackedSet<Channel>( new ConcurrentHashMap<Channel, Boolean>() );
	static final ConcurrentHashMap<Integer, Channel> connectingChannels = new ConcurrentHashMap<Integer, Channel>();
	
	protected static final String MT_PEER = "!_PEER";
	protected static final String MT_SERV = "!_SERV";
	protected static final String MT_HANDSHAKE = "!_HSM";
	protected static final String MT_HANDSHAKE_CONFIRM = "!_HSCONFIRM";
	protected static final String CMD_DISCONNECT = "!_BYE";	
	
	
	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelHandler#handleUpstream(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.ChannelEvent)
	 */
	@Override
	public void handleUpstream( ChannelHandlerContext ctx, ChannelEvent e ) throws Exception 
	{
		if ( e instanceof ChannelStateEvent )
		{
			logger.info ( e.toString() );
		}
		
		super.handleUpstream( ctx, e );
	}

	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelHandler#channelConnected(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.ChannelStateEvent)
	 */
	@Override
	public void channelConnected( ChannelHandlerContext ctx, ChannelStateEvent e ) throws Exception 
	{
		logger.debug( "Recieved connection request from " + e.getChannel().getRemoteAddress() );
		
		StringBuffer buffer = new StringBuffer( MT_HANDSHAKE + "|" );
		buffer.append( "cid:int:" + e.getChannel().getId() );		
		
		if( channels.size() != 0 )
		{
			buffer.append( "|peers:a:" );
			
			Iterator<Channel> i = channels.iterator();
			while ( i.hasNext() )
			{
				Channel c = i.next();
				buffer.append( c.getId() );				
				if( i.hasNext() )
				{
					buffer.append( "," );
				}			
			}
		}
		
		//channels.add( e.getChannel() );
		
		connectingChannels.put( e.getChannel().getId(), e.getChannel() );
		logger.debug( "Sending handshake message" );
		e.getChannel().write( buffer.toString() );
		
	}
	
	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelHandler#channelClosed(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.ChannelStateEvent)
	 */
	@Override
	public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception 
	{
		logger.debug( "Removing channel " + e.getChannel().getId() );
		channels.remove( e.getChannel() );
		
		for ( Channel c: channels )
		{
			c.write( MT_SERV + "|peer:int:" + e.getChannel().getId() );
		}		
		
	}

	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelHandler#messageReceived(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.MessageEvent)
	 */
	@Override
	public void messageReceived( ChannelHandlerContext ctx, MessageEvent e ) throws Exception 
	{
		String request = (String) e.getMessage();
		
		logger.debug( "MESSAGE RECIEVED [" + e.getChannel().getRemoteAddress() + "] " + request );
		
		if( request.indexOf( MT_HANDSHAKE_CONFIRM ) != -1  && connectingChannels.containsKey( e.getChannel().getId() ) )
		{
			confirmHandshake( e.getChannel() );
			return;
		}
		
		String response = MT_PEER + "|from:int:" + e.getChannel().getId() + "|msg:str:" + request;
		
		logger.debug( "Sending message: [ " + response + " ]");
		
		for ( Channel c: channels )
		{
			c.write( response );
		}		
	}

	/* (non-Javadoc)
	 * @see org.jboss.netty.channel.SimpleChannelHandler#exceptionCaught(org.jboss.netty.channel.ChannelHandlerContext, org.jboss.netty.channel.ExceptionEvent)
	 */
	@Override
	public void exceptionCaught( ChannelHandlerContext ctx, ExceptionEvent e ) throws Exception 
	{
		logger.warn( "Unexpected exception from downstream.", e.getCause() );
		
		e.getChannel().close();
		channels.remove( e.getChannel() );
	}		
	
	private void confirmHandshake( Channel channel )
	{	
		channels.add( connectingChannels.remove( channel.getId() ) );
		
		channel.write( "!_GO" );
		
		logger.debug( "Sending notify [ " + MT_SERV + "|peer:int:" + channel.getId() + " ]");
		
		Iterator<Channel> i = channels.iterator();
		
		while( i.hasNext() )
		{
			Channel c = i.next();
			
			if( c.getId() != channel.getId() )
			{
				c.write( MT_SERV + "|peer:int:" + channel.getId() );
			}
		}	
		
		logger.info( "Handshake confirmed for Channel " + channel.getId() );
		
	}
}





